package com.gitee.wsl.scheduler.sample


import com.gitee.wsl.coroutines.executor.KExecutor
import com.gitee.wsl.coroutines.executor.SuspendKExecutor
import com.gitee.wsl.concurrent.api.SuspendRunnable
import com.gitee.wsl.concurrent.api.SuspendRunnableWithResult
import com.gitee.wsl.coroutines.executor.sample.AsyncExecutor
import com.gitee.wsl.coroutines.executor.sample.AtomicExecutor
import com.gitee.wsl.coroutines.executor.sample.ConflatedExecutor
import com.gitee.wsl.coroutines.executor.sample.QueueExecutor
import com.gitee.wsl.scheduler.AbstractTaskScheduler
import com.gitee.wsl.scheduler.TaskScheduler
import com.gitee.wsl.scheduler.task.AsyncSchedulerTask
import com.gitee.wsl.scheduler.task.JobSchedulerTask
import com.gitee.wsl.scheduler.task.SchedulerTask
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Deferred
import kotlinx.coroutines.DelicateCoroutinesApi
import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.ObsoleteCoroutinesApi
import kotlinx.coroutines.Runnable
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.async
import kotlinx.coroutines.cancel
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.flow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onCompletion
import kotlinx.coroutines.flow.onEach
import kotlinx.coroutines.launch
import kotlin.coroutines.CoroutineContext
import kotlin.time.Duration


/**
 * # CoroutineScheduler
 *
 * CoroutineScheduler is a concrete implementation of [com.gitee.wsl.scheduler.AbstractTaskScheduler] that uses Kotlin coroutines
 * to schedule and run tasks.
 * It allows tasks to be run immediately, after a delay, or at fixed intervals.
 * This scheduler is suitable for applications that require a high degree of concurrency and efficiency.
 *
 * The tasks are managed within a [CoroutineScope], which ensures proper lifecycle management and
 * structured concurrency.
 *
 * ### Example usage:
 *
 * ```
 * val scheduler = CoroutineScheduler(CoroutineScope(Dispatchers.Default))
 *
 * // Run a task immediately
 * scheduler.runTask {
 *     println("Task running immediately")
 * }
 *
 * // Run a task after a 1000ms delay
 * scheduler.runTaskLater(1000L) {
 *     println("Task running after 1 second delay")
 * }
 *
 * // Run a task every 2000ms with an initial delay of 1000ms
 * scheduler.runTaskTimer(1000L, 2000L) {
 *     println("Task running every 2 seconds after an initial 1 second delay")
 * }
 * ```
 *
 * @property scope The [CoroutineScope] in which the tasks will be run.
 */
 open class CoroutineTaskScheduler(
     val scope: CoroutineScope
) : AbstractTaskScheduler(), CoroutineScope by scope {

    fun runTask(executor: KExecutor, task: Runnable): SchedulerTask {
        val taskId = nextId()
        val job = scope.launch {
            executor.execute {
                task.run()
                unregisterTask(taskId)
            }
        }
        return newTask(taskId, job)
    }

    fun runTask(executor: SuspendKExecutor, task: SuspendRunnable): SchedulerTask {
        val taskId = nextId()
        val job = scope.launch {
            executor.executeSuspend {
                task()
                unregisterTask(taskId)
            }
        }
        return newTask(taskId, job)
    }

    override fun runTask(task: SuspendRunnable): SchedulerTask {
        val taskId = nextId()
        val job = scope.launch {
            task()
            unregisterTask(taskId)
        }
        return newTask(taskId, job)
    }

    fun<T> asyncTask(task: SuspendRunnableWithResult<T>): AsyncSchedulerTask<T, Deferred<T>>{
        val taskId = nextId()
        val job = scope.async {
            val value = task()
            unregisterTask(taskId)
            value
        }
        return newTask(taskId, job)
    }


    @OptIn(ObsoleteCoroutinesApi::class)
    override fun runTaskTimer(
        delay: Duration,
        interval: Duration,
        task: SuspendRunnable
    ): SchedulerTask {
        val taskId = nextId()
        val flow = flow {
            delay(delay)
            emit(Unit)
            while (true) {
                delay(interval)
                emit(Unit)
            }
        }.onEach {
                scope.launch {
                    task()
                }
            }
            .onCompletion { unregisterTask(taskId) }
            .launchIn(scope)

        return newTask(taskId, flow)
    }

    override fun runTaskLater(
        delay: Duration,
        task: SuspendRunnable
    ): SchedulerTask {
        val taskId = nextId()

        val job = scope.launch {
            delay(delay)
            task()
        }.also {
            it.invokeOnCompletion {
                unregisterTask(taskId)
            }
        }

        return newTask(taskId, job)
    }

    fun <JOB: Job> newTask(id: Int, job: JOB): JobSchedulerTask<JOB> {
        val task = JobSchedulerTaskImp(id = id, job = job, finalBlock = {unregisterTask(id)})

        registerTask(task)
        return task
    }

    fun <T, D: Deferred<T>> newTask(id: Int, job: D): AsyncSchedulerTask<T,D> {
        val task = object : AsyncSchedulerTask<T,D> {

            override val job: D = job

            override fun cancel() {
                try {
                    job.cancel()
                } finally {
                    unregisterTask(id)
                }
            }
            override val id: Int = id
        }
        registerTask(task)
        return task
    }


    override fun shutdown() {
        cancelAllTasks()
        scope.cancel()
    }

}

/**
 * Creates a [CoroutineTaskScheduler] with a [SupervisorJob] in the provided [CoroutineContext].
 *
 * @param context The [CoroutineContext] to be used for the [CoroutineScope].
 * @return A new instance of [CoroutineTaskScheduler].
 */

fun TaskScheduler.Companion.fromSupervisor(context: CoroutineContext): CoroutineTaskScheduler =
         CoroutineTaskScheduler(CoroutineScope(context + SupervisorJob()))

/**
 * A globally available [CoroutineTaskScheduler] using the [GlobalScope].
 * This should be used with caution as tasks in the GlobalScope are not bound to any specific lifecycle.
 */
@DelicateCoroutinesApi
val TaskScheduler.Companion.Global: CoroutineTaskScheduler by lazy {
    CoroutineTaskScheduler(GlobalScope)
}

fun CoroutineTaskScheduler.queue(task: SuspendRunnable): SchedulerTask = runTask(QueueExecutor.INSTANCE,task)

fun CoroutineTaskScheduler.blocking(task: SuspendRunnable): SchedulerTask = runTask(AsyncExecutor.INSTANCE,task)

fun CoroutineTaskScheduler.atomic(task: () -> Unit): SchedulerTask = runTask(AtomicExecutor.INSTANCE,task)

fun CoroutineTaskScheduler.conflated(task: SuspendRunnable): SchedulerTask = runTask(ConflatedExecutor.INSTANCE,task)

internal class JobSchedulerTaskImp<JOB: Job>(
      override val id: Int,
      override val job: JOB,
      val finalBlock:()->Unit
    ):JobSchedulerTask<JOB>{

    override fun cancel() {
        try {
            job.cancel()
        } finally {
            finalBlock()
        }
    }
}